package cn.mesmile.flink.demo;

import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.java.DataSet;
import org.apache.flink.api.java.ExecutionEnvironment;
import org.apache.flink.util.Collector;

/**
 * @author zb
 * @date 2021/8/21 11:26
 * @Description  推荐使用 {@link FlinkStreamApp01}
 */
public class FlinkDataSetApp02 {

    /**
     *  批处理
     *      Flink1.12时支持流批一体，DataSetAPI已经不推荐使用了，案例都会优先使用DataStream流式API
     *      {@link FlinkStreamApp01}
     * */
    public static void main(String[] args) throws Exception {
        // 构建执行任务环境以及任务的启动的入口, 存储全局相关的参数
        ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
        //  设置并行度
        env.setParallelism(3);
        // 相同类型元素的数据集 source
        DataSet<String> stringDS = env.fromElements("java,SpringBoot", "spring cloud,redis", "kafka,class");

        stringDS.printOnTaskManager("处理前");
        // FlatMapFunction<String, String>, key是输入类型，value是Collector响应的收集的类型，看源码注释，也是 DataStream<String>里面泛型类型
        DataSet<String> flatMapDS = stringDS.flatMap(new FlatMapFunction<String, String>() {
            @Override
            public void flatMap(String value, Collector<String> collector) throws Exception {
                String [] arr =  value.split(",");
                for(String str : arr){
                    collector.collect(str);
                }
            }
        });
        //输出 sink
        flatMapDS.printOnTaskManager("处理后");
        //DataStream需要调用execute,可以取个名称
        env.execute("flat map job");
    }
}
